package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.app.func.BeanToJsonStrFunction;
import realtime.app.func.DimAsyncFunction;
import realtime.beans.TrafficSidScBean;
import realtime.util.DateFormatUtil;
import realtime.util.DorisUtil;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

import java.util.Collections;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.TimeUnit;

public class DwsTrafficScWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_sc_window_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        //转化为JSONObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject jsonObj = JSON.parseObject(jsonStr);
                    out.collect(jsonObj);
            }
        });
        //指定水位线
        SingleOutputStreamOperator<JSONObject> watermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("ts");
                            }
                        })
        );
        //按照mid分组
        KeyedStream<JSONObject, String> keyedDS = watermarkDS.keyBy(v1 -> v1.getJSONObject("common").getString("mid"));
        //对数据处理并且转换为实体类
        SingleOutputStreamOperator<TrafficSidScBean> beanDS = keyedDS.process(
                new KeyedProcessFunction<String, JSONObject, TrafficSidScBean>() {
                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("midValueState", String.class);
                        valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                        lastVisitDateState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, TrafficSidScBean>.Context ctx, Collector<TrafficSidScBean> out) throws Exception {
                        JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                        String uid = commonJsonObj.getString("uid");
                        String sid = commonJsonObj.getString("sid");
                        String sc = commonJsonObj.getString("sc");
                        String isNew = commonJsonObj.getString("is_new");
                        Long ts = jsonObj.getLong("ts");
                        Long durTime = jsonObj.getJSONObject("page").getLong("during_time");
                        String curDate = DateFormatUtil.toDate(ts);
                        String lastVisitDate = lastVisitDateState.value();
                        Long uvCt = 0L;
                        //如果上次访问时间是空 或者 不是今天的日期 说明是今天第一次访问 为独立访客
                        if (StringUtils.isEmpty(lastVisitDate) || !lastVisitDate.equals(curDate)) {
                            uvCt = 1L;
                            lastVisitDateState.update(curDate);
                        }
                        String lagePageId = jsonObj.getJSONObject("page").getString("last_page_id");
                        Long sidCt = 0L;
                        //如果上一个页面为空 说明是新会话
                        if (StringUtils.isEmpty(lagePageId)) {
                            sidCt = 1L;
                        }
                        TrafficSidScBean trafficSicSC = TrafficSidScBean.builder()
                                .userId(uid)
                                .sid(sid)
                                .sc(sc)
                                .isNew(isNew)
                                .uvCt(uvCt)
                                .sidCt(sidCt)
                                .pageCt(1L)
                                .durSum(durTime)
                                .build();
                        out.collect(trafficSicSC);
                    }
                }
        );
        //按照会话分组
        KeyedStream<TrafficSidScBean, String> sidKeyedDS = beanDS.keyBy(
                v1 -> v1.getSid()
        );
        //开窗
        WindowedStream<TrafficSidScBean, String, TimeWindow> windowDS = sidKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));
        //聚合
        SingleOutputStreamOperator<TrafficSidScBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TrafficSidScBean>() {
                    @Override
                    public TrafficSidScBean reduce(TrafficSidScBean v1, TrafficSidScBean v2) throws Exception {
                        v1.setPageCt(v1.getPageCt() + v2.getPageCt());
                        v1.setDurSum(v1.getDurSum() + v2.getDurSum());
                        return v1;
                    }
                },
                new WindowFunction<TrafficSidScBean, TrafficSidScBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TrafficSidScBean> input, Collector<TrafficSidScBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        long timeMillis = System.currentTimeMillis();
                        String curDate = DateFormatUtil.toDate(timeMillis);
                        for (TrafficSidScBean trafficSidScBean : input) {
                            trafficSidScBean.setStt(stt);
                            trafficSidScBean.setEdt(edt);
                            trafficSidScBean.setCurDate(curDate);
                            out.collect(trafficSidScBean);
                        }
                    }
                }
        );
        //关联维度表dim_base_source  补充维度字段 source_site
        SingleOutputStreamOperator<TrafficSidScBean> resultDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<TrafficSidScBean>("dim_base_source") {
                    @Override
                    public String getKey(TrafficSidScBean bean) {
                        return bean.getSc();
                    }

                    @Override
                    public void join(TrafficSidScBean bean, JSONObject dim) {
                        bean.setSource_site(dim.getString("source_site"));
                    }
                },
                60,
                TimeUnit.SECONDS
        );


        //resultDS.print(">>>>>>>>>>");

        resultDS.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_traffic_source_window"));
        env.execute();
    }
}
